package com.bujunjie.study.realtime.common.util;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * <p>HBase 的工具类</p>
 *
 * @author bu.junjie
 * @version 1.0.0
 * @createTime 2025/9/12 17:26
 */
@Slf4j
public class HBaseUtil {

    /**
     * <p>获取Hbase连接</p>
     *
     * @param
     * @return 返回 Hbase连接
     * @author bu.junjie
     * @date 2025/9/12 17:27
     */
    public static Connection getHBaseConnection() throws IOException {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");

        Connection hbaseConn = ConnectionFactory.createConnection(conf);
        return hbaseConn;
    }

    /**
     * <p>关闭Hbase连接</p>
     *
     * @param hbaseConn hbase 连接
     * @return
     * @author bu.junjie
     * @date 2025/9/12 17:27
     */
    public static void closeHBaseConnection(Connection hbaseConn) throws IOException {
        if (hbaseConn != null && !hbaseConn.isClosed()) {
            hbaseConn.close();
        }
    }

    /**
     * <p></p>
     *
     * @param hbaseConn 连接
     * @param namespace 命名空间
     * @param tableName 表名
     * @param hbaseConn 连接
     * @return
     * @author bu.junjie
     * @date 2025/9/12 17:42
     */
    public static void createHBaseTable(Connection hbaseConn, String namespace, String tableName, List<String> families) {
        if (CollectionUtil.isEmpty(families)) {
            log.error("至少需要一个列族");
            return;
        }

        try (Admin admin = hbaseConn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            if (admin.tableExists(tableNameObj)) {
                log.info("表空间=【{}】,下的表tableName=【{}】已存在", namespace, tableName);
                return;
            }
            TableDescriptorBuilder tableDescriptorBuilder = TableDescriptorBuilder.newBuilder(tableNameObj);
            for (String family : families) {
                ColumnFamilyDescriptor columnFamilyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(Bytes.toBytes(family)).build();
                tableDescriptorBuilder.setColumnFamily(columnFamilyDescriptor);
            }

            admin.createTable(tableDescriptorBuilder.build());
            log.info("表空间=【{}】,下的表tableName=【{}】创建成功", namespace, tableName);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * <p>删除 hbase 中的列名</p>
     *
     * @param hbaseConn 连接
     * @param namespace 命名空间
     * @param tableName 表名
     * @return
     * @author bu.junjie
     * @date 2025/9/12 17:37
     */
    public static void dropHBaseTable(Connection hbaseConn, String namespace, String tableName) {

        try (Admin admin = hbaseConn.getAdmin()) {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            //判断要删除的表是否存在
            if (!admin.tableExists(tableNameObj)) {
                log.info("要删除的表空间=【{}】,下的表tableName=【{}】不存在", namespace, tableName);
                return;
            }
            admin.disableTable(tableNameObj);
            admin.deleteTable(tableNameObj);
            log.info("删除的表空间=【{}】,下的表tableName=【{}】", namespace, tableName);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * <p>删除行数据</p>
     *
     * @param hbaseConn hbase 连接
     * @param namespace 命名空间
     * @param tableName 表名
     * @param rowKey    表的主键值
     * @return
     * @author bu.junjie
     * @date 2025/9/15 9:40
     */
    public static void delRow(Connection hbaseConn, String namespace, String tableName, String rowKey) {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)) {
            Delete delete = new Delete(Bytes.toBytes(rowKey));
            table.delete(delete);
            log.info("从表空间=【{}】,下的表tableName=【{}】中删除数据rowKey=【{}】成功", namespace, tableName, rowKey);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * 向表中put数据
     *
     * @param hbaseConn 连接对象
     * @param namespace 表空间
     * @param tableName 表名
     * @param rowKey    rowkey
     * @param family    列族
     * @param dataMap   要put的数据
     */
    public static void putRow(Connection hbaseConn, String namespace, String tableName, String rowKey, String family, Map<String, Object> dataMap) {
        TableName tableNameObj = TableName.valueOf(namespace, tableName);
        try (Table table = hbaseConn.getTable(tableNameObj)) {
            Put put = new Put(Bytes.toBytes(rowKey));
            Set<String> columns = dataMap.keySet();
            for (String column : columns) {
                String value = (String) dataMap.get(column);
                if (StrUtil.isNotBlank(value)) {
                    put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
                }
            }
            table.put(put);
            log.info("向表空间=【{}】,下的表tableName=【{}】中put数据 rowKey=【{}】成功", namespace, tableName, rowKey);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * <p>获取异步操作HBase的连接对象</p>
     *
     * @return 返回异步的连接池对象
     * @author bu.junjie
     * @date 2025/9/18 17:15
     */
    public static AsyncConnection getHBaseAsyncConnection() {
        Configuration conf = new Configuration();
        conf.set("hbase.zookeeper.quorum", "hadoop102,hadoop103,hadoop104");

        try {
            return ConnectionFactory.createAsyncConnection(conf).get();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    /**
     * <p>以异步的方式 从HBase维度表中查询维度数据</p>
     *
     * @param asyncConn 异步操作HBase的连接
     * @param namespace 表空间
     * @param tableName 表名
     * @param rowKey    rowkey
     * @return 返回 HBase 表中的数据
     * @author bu.junjie
     * @date 2025/9/18 17:32
     */
    public static JSONObject readDimAsync(AsyncConnection asyncConn, String namespace, String tableName, String rowKey) {
        try {
            TableName tableNameObj = TableName.valueOf(namespace, tableName);
            AsyncTable<AdvancedScanResultConsumer> asyncTable = asyncConn.getTable(tableNameObj);
            Get get = new Get(Bytes.toBytes(rowKey));
            Result result = asyncTable.get(get).get();
            List<Cell> cells = result.listCells();
            if (cells != null && cells.size() > 0) {
                JSONObject jsonObj = new JSONObject();
                for (Cell cell : cells) {
                    String columnName = Bytes.toString(CellUtil.cloneQualifier(cell));
                    String columnValue = Bytes.toString(CellUtil.cloneValue(cell));
                    jsonObj.put(columnName, columnValue);
                }
                return jsonObj;
            }
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
        return null;
    }


    /**
     * <p>关闭异步操作HBase的连接对象</p>
     *
     * @param asyncConn 异步连接池
     * @author bu.junjie
     * @date 2025/9/18 17:16
     */
    public static void closeAsyncHbaseConnection(AsyncConnection asyncConn) {
        if (asyncConn != null && !asyncConn.isClosed()) {
            try {
                asyncConn.close();
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }
}



